package org.example.flink.join;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.example.flink.source.S1;
import org.example.flink.source.S2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;
import scala.Tuple3;

import java.time.LocalTime;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * Flink 双流JOIN ，s1 join s2 。 相同KEY，VALUE表示内连接(如果相同key，s1有m个value，s2有n个value，则输出 m * n 条记录)
 */
public class FJoin {

    private static Logger logger = LoggerFactory.getLogger(FJoin.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Tuple3<String, Integer, Integer>> s1 = env.addSource(new S1());
        DataStreamSource<Tuple3<String, Integer, Integer>> s2 = env.addSource(new S2());

        DataStream<String> apply = s1.join(s2).where(new KeySelector<Tuple3<String, Integer, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer, Integer> v1) throws Exception {
                return v1._1();
            }
        }).equalTo(new KeySelector<Tuple3<String, Integer, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer, Integer> v2) throws Exception {
                return v2._1();
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
            .apply(new JoinFunction<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>, String>() {
                @Override
                public String join(Tuple3<String, Integer, Integer> v1, Tuple3<String, Integer, Integer> v2)
                    throws Exception {
                    String s = new Date() + " " + v1._1() + " (" + v1._2() + "," + v1._3() + ") (" + v2._2() + ","
                        + v2._3() + ")";
                    logger.info(s);
                    return s;
                }
            });

        apply.print();

        env.execute("FJoin");

    }

}
